package com.lzy.elasticserch.config;

import com.alibaba.otter.canal.client.CanalConnector;
import com.lzy.canal.config.CanalClientAutoConfig;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.ToString;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Map;
import java.util.stream.Collectors;

/**
 * canal配置类
 */
@Configuration
@EnableConfigurationProperties({CanalConsumerProperty.class})
public class CanalConsumerConfig {


    @Bean
    @ConditionalOnMissingBean
    public CanalConsumerManage getCanalConsumerManage(CanalConsumerProperty canalConsumerProperty, CanalClientAutoConfig.CanalClientConnectorManage canalClientConnectorManage) {
        return new CanalConsumerManage(canalConsumerProperty, canalClientConnectorManage);
    }

    /**
     * canal consumer管理器
     */
    public static class CanalConsumerManage {
        private final Map<String, CanalConsumerClient> map;

        public CanalConsumerManage(CanalConsumerProperty canalConsumerProperty, CanalClientAutoConfig.CanalClientConnectorManage canalClientConnectorManage) {
            map = canalConsumerProperty.getCanalConsumerItems().stream()
                    .collect(Collectors.toMap(CanalConsumerProperty.CanalConsumerItem::getCode, o -> {
                        CanalConsumerClient client = new CanalConsumerClient();
                        client.setCanalConsumerItemProperty(o);
                        client.setCanalConnector(canalClientConnectorManage.get(o.getDestination()));
                        return client;
                    }));

        }

        public CanalConsumerClient getCanalConsumerResourceClient(String code) {
            return map.get(code);
        }
    }


    @Data
    @ToString
    @EqualsAndHashCode
    public static class CanalConsumerClient {
        /**
         * 连接对象
         */
        private CanalConnector canalConnector;
        /**
         * 消费配置
         */
        private CanalConsumerProperty.CanalConsumerItem canalConsumerItemProperty;
    }


}
